package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.GmallConfig;
import com.atguigu.gmall.realtime.utils.DimUtil;
import com.atguigu.gmall.realtime.utils.PhoenixUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.util.Collection;
import java.util.Set;

/**
 * @author Felix
 * @date 2022/11/3
 * 将流中的数据写到phoenix表中
 */
public class DimSinkFunction implements SinkFunction<JSONObject> {
    @Override
    public void invoke(JSONObject jsonObj, Context context) throws Exception {
        //获取输出目的地
        String sinkTable = jsonObj.getString("sink_table");
        //获取到输出目的地之后，将sink_table属性从json对象中移除掉  {"tm_name":"八星星","id":12}
        jsonObj.remove("sink_table");

        String type = jsonObj.getString("type");
        jsonObj.remove("type");

        //获取json中所有属性，作为向数据库中插入数据字段的名称
        Set<String> keys = jsonObj.keySet();
        //获取json中所有属性的值，作为向数据库中插入数据字段的值
        Collection<Object> values = jsonObj.values();
        //拼接插入语句
        String upsertSql = "upsert into "+ GmallConfig.PHOENIX_SCHEMA +"."+sinkTable
            +" ("+ StringUtils.join(keys,",") +") values ('"+StringUtils.join(values,"','")+"')";
        System.out.println("向phoenix表中插入数据的SQL:" +upsertSql);

        //如果对Phoenix的维度数据进行了修改，将Redis中缓存的维度删除掉
        if("update".equals(type)){
            DimUtil.delCached(sinkTable,jsonObj.getString("id"));
        }

        PhoenixUtil.executeSql(upsertSql);
    }
}
